package com.demo.common.redis.delay;

import lombok.extern.slf4j.Slf4j;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.Cancelable;
import org.springframework.data.redis.stream.Task;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.function.Function;

/**
 * <p>
 * Redis的延迟消息容器。
 * </p>
 *
 * @author molong
 * @date 2022/6/22
 */
@Slf4j
public class DefaultDelayMessageListenerContainer implements DelayMessageListenerContainer {

    private final Object lifecycleMonitor = new Object();
    private final Executor taskExecutor;
    private final RedisTemplate<String, DelayMessage> template;
    private final List<Subscription> subscriptions = new ArrayList<>();
    /**
     * 一次拉取多少条数据
     */
    private long batchSize = 10;

    private boolean running = false;

    /**
     * Create a new DefaultDelayMessageListenerContainer
     *
     * @param connectionFactory must not be {@literal null}.
     */
    DefaultDelayMessageListenerContainer(RedisConnectionFactory connectionFactory) {

        Assert.notNull(connectionFactory, "RedisConnectionFactory must not be null!");
        this.taskExecutor = new SimpleAsyncTaskExecutor();
        this.template = createRedisTemplate(connectionFactory);
    }

    private RedisTemplate<String, DelayMessage> createRedisTemplate(RedisConnectionFactory connectionFactory) {

        RedisTemplate<String, DelayMessage> redisTemplate = new RedisTemplate<>();
        redisTemplate.setKeySerializer(StringRedisSerializer.UTF_8);
        redisTemplate.setConnectionFactory(connectionFactory);
        redisTemplate.afterPropertiesSet();

        return redisTemplate;
    }

    @Override
    public void setRedisSerializer(RedisSerializer<?> redisSerializer) {
        this.template.setValueSerializer(redisSerializer);
    }

    @Override
    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    @Override
    public boolean isAutoStartup() {
        return false;
    }


    @Override
    public void stop(Runnable callback) {

        stop();
        callback.run();
    }

    @Override
    public void start() {

        synchronized (lifecycleMonitor) {

            if (this.running) {
                return;
            }

            subscriptions.stream()
                    //过滤掉正在运行的订阅
                    .filter(it -> !it.isActive())
                    //过滤掉为当前订阅实现的
                    .filter(it -> it instanceof TaskSubscription)
                    //启动任务
                    .map(TaskSubscription.class::cast)
                    .map(TaskSubscription::getTask)
                    .forEach(taskExecutor::execute);

            running = true;
        }
    }

    @Override
    public void stop() {

        synchronized (lifecycleMonitor) {

            if (this.running) {

                subscriptions.forEach(Cancelable::cancel);

                running = false;
            }
        }
    }

    @Override
    public boolean isRunning() {

        synchronized (this.lifecycleMonitor) {
            return running;
        }
    }

    @Override
    public int getPhase() {
        return Integer.MAX_VALUE;
    }

    @Override
    public Subscription register(DelayMessageListener listener) {
        return doRegister(getReadTask(listener));
    }

    private DelayPollTask getReadTask(DelayMessageListener listener) {
        Function<String, List<ZSetOperations.TypedTuple<DelayMessage>>> readFunction = getReadFunction(listener.getRedisKey());
        return new DelayPollTask(listener, readFunction);
    }

    private Function<String, List<ZSetOperations.TypedTuple<DelayMessage>>> getReadFunction(String key) {
        return redisKey -> {
            //先查询是否有超过当前时间的消息。
            Long count = template.opsForZSet().count(key, ScoreUtil.getScore(ScoreUtil.START_TIME), ScoreUtil.getScore(LocalDateTime.now()));
            if(Objects.isNull(count) || count <= 0){
                return null;
            }
            //存在数据，拉取数据
            Set<? extends ZSetOperations.TypedTuple<DelayMessage>> typedTuples = template.opsForZSet().rangeWithScores(key, 0, batchSize - 1);
            if(CollectionUtils.isEmpty(typedTuples)){
                return null;
            }
            //删除数据，表示已经被这任务获取到，没有删除成功的，代表被其他的task获取到
            List<ZSetOperations.TypedTuple<DelayMessage>> receiveMsg = new ArrayList<>();
            for (ZSetOperations.TypedTuple<DelayMessage> typedTuple : typedTuples) {
                Long remove = template.opsForZSet().remove(key, typedTuple.getValue());
                if(Objects.nonNull(remove) && remove > 0){
                    receiveMsg.add(typedTuple);
                }
            }
            return receiveMsg;
        };
    }

    private Subscription doRegister(Task task) {

        Subscription subscription = new TaskSubscription(task);

        synchronized (lifecycleMonitor) {

            this.subscriptions.add(subscription);

            if (this.running) {
                taskExecutor.execute(task);
            }
        }

        return subscription;
    }

    @Override
    public void remove(Subscription subscription) {

        synchronized (lifecycleMonitor) {

            if (subscriptions.contains(subscription)) {

                if (subscription.isActive()) {
                    subscription.cancel();
                }

                subscriptions.remove(subscription);
            }
        }
    }

    /**
     * 基于Task的订阅实现
     */
    static class TaskSubscription implements Subscription {

        private final Task task;

        protected TaskSubscription(Task task) {
            this.task = task;
        }

        Task getTask() {
            return task;
        }

        @Override
        public boolean isActive() {
            return task.isActive();
        }

        @Override
        public boolean await(Duration timeout) throws InterruptedException {
            return task.awaitStart(timeout);
        }

        @Override
        public void cancel() throws DataAccessResourceFailureException {
            task.cancel();
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }

            TaskSubscription that = (TaskSubscription) o;

            return ObjectUtils.nullSafeEquals(task, that.task);
        }

        @Override
        public int hashCode() {
            return ObjectUtils.nullSafeHashCode(task);
        }
    }
}